/**
 * 
 */
package com.redis.spi;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;

import com.redis.api.DataReader;
import com.redis.bean.KeyValue;
import com.redis.io.utils.HessianSerializerUtils;

/**
 * @author yangjuanying
 */
public class JedisDataReader implements DataReader {

	/**
	 * jedis只读客户端连接池
	 */
	private final ShardedJedisPool shardedJedisPool;
	
	public JedisDataReader(final ShardedJedisPool readOnlyPool){
		this.shardedJedisPool = readOnlyPool;
	}
	
	/**
	 * 回收jedis客户端
	 * @param isNormal                 是否正常
	 * @param jedis                    jedis
	 */
	private void release(boolean isNormal, ShardedJedis jedis){
		if (isNormal)
			this.shardedJedisPool.returnResource(jedis);
		else {
			this.shardedJedisPool.returnBrokenResource(jedis);
		}
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#get(byte[])
	 */
	public byte[] get(byte[] key) {
		return get(key, null);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#get(java.lang.String)
	 */
	public String get(String key) {
		return get(key, null);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#getBean(java.lang.String)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T getBean(String key) {
		return (T)HessianSerializerUtils.deSerialize(get(key.getBytes()));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#get(byte[], byte[])
	 */
	public byte[] get(byte[] key, byte[] emptyDefault) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		byte[] value = null;
		try {
			value = jedis.get(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null?emptyDefault : value;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#getBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> T getBean(String key, T t) {
		T v = getBean(key);
		return v == null ? t : v;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#get(java.lang.String, java.lang.String)
	 */
	public String get(String key, String emptyDefault) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String value = null;
		try {
			value = jedis.get(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null? emptyDefault : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hget(byte[], byte[])
	 */
	public byte[] hget(byte[] key, byte[] field) {
		return hget(key, field, null);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hget(byte[], byte[], byte[])
	 */
	public byte[] hget(byte[] key, byte[] field, byte[] emptyDefault) {
		byte[] value = null;
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			value = jedis.hget(key, field);
			flag = true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null?emptyDefault : value;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hget(java.lang.String, java.lang.String)
	 */
	public String hget(String key, String field) {
		return hget(key, field, null);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hget(java.lang.String, java.lang.String, java.lang.String)
	 */
	public String hget(String key, String field, String emptyDefault) {
		String value = null;
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			value = jedis.hget(key, field);
			flag = true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null? emptyDefault : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hgetBean(java.lang.String, java.lang.String)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T hgetBean(String key, String field) {
		return (T)HessianSerializerUtils.deSerialize(hget(key.getBytes(), field.getBytes()));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hgetBean(java.lang.String, java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> T hgetBean(String key, String field,
			T emptyDefault) {
		T v = hgetBean(key, field);
		return v == null ? emptyDefault : v;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hgetAll(java.lang.String)
	 */
	public List<KeyValue> hgetAll(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		List<KeyValue> kvs = null;
		boolean flag = false;
		try {
			Map<byte[], byte[]> values = jedis.hgetAll(key.getBytes());
			if(values != null && !values.isEmpty()){
				kvs = new ArrayList<KeyValue>();
				Iterator<Entry<byte[], byte[]>> it = values.entrySet().iterator();
				while(it.hasNext()){
					Entry<byte[], byte[]> entry = it.next();
					KeyValue kv = new KeyValue(new String(entry.getKey()), entry.getValue());
					kvs.add(kv);
				}
			}
			flag = true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return kvs;
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private List keys(String pattern, boolean isBinary){
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		List keys = null;
		try {
			Jedis j = jedis.getShard(pattern);
			Set set = null;
			if(isBinary){
				set = j.keys(pattern.getBytes());
			}else{
				set = j.keys(pattern);
			}
			if(set != null && !set.isEmpty()){
				keys = new ArrayList();
				for(Object o : set){
					keys.add(o);
				}
			}
			flag = true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return keys;
	}
	

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#keys(java.lang.String)
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public String[] keys(String pattern) {
		List keys = keys(pattern, false);
		if(keys != null && keys.size() > 0){
			String[] ks = new String[keys.size()];
			keys.toArray(ks);
			return ks;
		}else{
			return null;
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hkeys(java.lang.String)
	 */
	public String[] hkeys(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String[] keys = null;
		try {
			Jedis j = jedis.getShard(key);
			Set<String> set = j.hkeys(key);
			if(set != null && !set.isEmpty()){
				keys = new String[set.size()];
				set.toArray(keys);
			}
			flag = true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return keys;
	}


	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#values(java.lang.String)
	 */
	public List<KeyValue> values(String pattern) {
		String[] keys = keys(pattern);
		List<KeyValue> kvs = null;
		if(keys != null && keys.length > 0){
			ShardedJedis jedis = this.shardedJedisPool.getResource();
			boolean flag = false;
			byte[][] byteKeys = new byte[keys.length][];
			try{
				Jedis j = jedis.getShard(pattern);
				List<byte[]> values = j.mget(byteKeys);
				if(values != null && values.size() > 0){
					kvs = new ArrayList<KeyValue>();
					for(int i=0;i<values.size(); i++){
						KeyValue kv = new KeyValue(keys[i], values.get(i));
						kvs.add(kv);
					}
				}
				flag = true;
			} catch(Exception e){
				e.printStackTrace();
				flag = false;
			} finally {
				release(flag, jedis);
			}
		}
		return kvs;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#ttl(java.lang.String)
	 */
	public long ttl(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.ttl(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? -2 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#pttl(java.lang.String)
	 */
	public long pttl(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			Jedis j = jedis.getShard(key);
			value = j.pttl(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? -2 : value;
	}


	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#type(java.lang.String)
	 */
	public String type(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String type = null;
		try {
			type = jedis.type(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return type == null ? "none" : type;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#exists(java.lang.String)
	 */
	public boolean exists(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Boolean exists = null;
		try {
			exists = jedis.exists(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return exists == null ? false : exists;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#getRange(java.lang.String, int, int)
	 */
	public String getRange(String key, int start, int end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String subString = null;
		try {
			subString = jedis.getrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return subString;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hlen(java.lang.String)
	 */
	public long hlen(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long count = new Long(0);
		try {
			count = jedis.hlen(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return count == null ? 0 : count.longValue();
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#mget(java.lang.String[])
	 */
	public List<KeyValue> mget(String... key) {
		List<KeyValue> kvs = null;
		if(key != null && key.length > 0){
			ShardedJedis jedis = this.shardedJedisPool.getResource();
			boolean flag = false;
			try {
				Jedis j = jedis.getShard(key[0]);
				byte[][] keys = new byte[key.length][];
				for(int i=0;i<key.length;i++){
					keys[i] = key[i].getBytes();
				}
				List<byte[]> values = j.mget(keys);
				if(values != null && values.size() > 0){
					kvs = new ArrayList<KeyValue>();
					for(int i=0;i<values.size();i++){
						KeyValue kv = new KeyValue(key[i], values.get(i));
						kvs.add(kv);
					}
				}
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return kvs;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#mget(java.util.List)
	 */
	public List<KeyValue> mget(List<String> keys) {
		if(keys !=null && keys.size() >0){
			String[] tmps = new String[keys.size()];
			keys.toArray(tmps);
			return mget(tmps);
		}else{
			return null;
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#hexists(java.lang.String, java.lang.String)
	 */
	public boolean hexists(String key, String field) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Boolean exists = null;
		try {
			exists = jedis.hexists(key, field);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return exists == null ? false: exists;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#lindex(java.lang.String, long)
	 */
	public byte[] lindex(String key, long index) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		byte[] values = null;
		try {
			values = jedis.lindex(key.getBytes(), index);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return values;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#lrange(byte[], long, long)
	 */
	public List<byte[]> lrange(byte[] key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		List<byte[]> values = null;
		try {
			values = jedis.lrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return values;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#lrange(java.lang.String, long, long)
	 */
	public List<String> lrange(String key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		List<String> values = null;
		try {
			values = jedis.lrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return values;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#lrangeBean(byte[], long, long)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> List<T> lrangeBean(byte[] key, long start,
			long end) {
		List<byte[]> values = lrange(key, start, end);
		List<T> result = null;
		if(values != null){
			result = new ArrayList<T>();
			for(byte[] v : values){
				result.add((T)HessianSerializerUtils.deSerialize(v));
			}
		}
		return result;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#llen(java.lang.String)
	 */
	public long llen(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.llen(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#scard(java.lang.String)
	 */
	public long scard(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.scard(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sismember(java.lang.String, byte[])
	 */
	public boolean sismember(String key, byte[] member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Boolean value = null;
		try {
			value = jedis.sismember(key.getBytes(), member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? false : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sismember(java.lang.String, java.lang.String)
	 */
	public boolean sismember(String key, String member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Boolean value = null;
		try {
			value = jedis.sismember(key, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? false : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sismember(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> boolean sismember(String key, T member) {
		return sismember(key, HessianSerializerUtils.serialize(member));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#smembers(java.lang.String)
	 */
	public Set<byte[]> smembers(byte[] key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		try {
			value = jedis.smembers(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#smembers(java.lang.String)
	 */
	public Set<String> smembers(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		try {
			value = jedis.smembers(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#smembersBean(byte[])
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> smembersBean(byte[] key) {
		Set<byte[]> values = smembers(key);
		Set<T>  result = null;
		if(values != null){
			result = new HashSet<T>();
			for(byte[] v : values){
				result.add((T)HessianSerializerUtils.deSerialize(v));
			}
		}
		return result;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sunion(byte[][])
	 */
	public Set<byte[]> sunion(byte[]... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sunion(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sunion(java.lang.String[])
	 */
	public Set<String> sunion(String... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sunion(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sunionBean(byte[][])
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> sunionBean(byte[]... keys) {
		Set<byte[]> values = sunion(keys);
		Set<T>  result = null;
		if(values != null){
			result = new HashSet<T>();
			for(byte[] v : values){
				result.add((T)HessianSerializerUtils.deSerialize(v));
			}
		}
		return result;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sdiff(byte[][])
	 */
	public Set<byte[]> sdiff(byte[]... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sdiff(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sdiff(java.lang.String[])
	 */
	public Set<String> sdiff(String... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sdiff(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sdiffBean(byte[][])
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> sdiffBean(byte[]... keys) {
		Set<byte[]> values = sdiff(keys);
		Set<T>  result = null;
		if(values != null){
			result = new HashSet<T>();
			for(byte[] v : values){
				result.add((T)HessianSerializerUtils.deSerialize(v));
			}
		}
		return result;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sinter(byte[][])
	 */
	public Set<byte[]> sinter(byte[]... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sinter(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sinter(java.lang.String[])
	 */
	public Set<String> sinter(String... keys) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		if(keys !=null && keys.length > 0){
			try {
				Jedis j = jedis.getShard(keys[0]);
				value = j.sinter(keys);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#sinterBean(byte[][])
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> sinterBean(byte[]... keys) {
		Set<byte[]> values = sinter(keys);
		Set<T>  result = null;
		if(values != null){
			result = new HashSet<T>();
			for(byte[] v : values){
				result.add((T)HessianSerializerUtils.deSerialize(v));
			}
		}
		return result;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zcard(java.lang.String)
	 */
	public long zcard(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zcard(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zcount(java.lang.String, double, double)
	 */
	public long zcount(String key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zcount(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrange(byte[], long, long)
	 */
	public Set<byte[]> zrange(byte[] key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		try {
			value = jedis.zrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrange(java.lang.String, long, long)
	 */
	public Set<String> zrange(String key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		try {
			value = jedis.zrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrangeBean(java.lang.String, long, long)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> zrangeBean(String key, long start,
			long end) {
		Set<byte[]> values = zrange(key.getBytes(), start, end);
		Set<T> ret = null;
		if(values != null && !values.isEmpty()){
			ret = new HashSet<T>();
			for(byte[] value : values){
				ret.add((T)HessianSerializerUtils.deSerialize(value));
			}
			return ret;
		}else{
			return null;   
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrangeByScore(byte[], double, double)
	 */
	public Set<byte[]> zrangeByScore(byte[] key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		try {
			value = jedis.zrangeByScore(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrangeByScore(java.lang.String, double, double)
	 */
	public Set<String> zrangeByScore(String key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		try {
			value = jedis.zrangeByScore(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrangeByScoreBean(java.lang.String, double, double)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> zrangeByScoreBean(String key,
			double min, double max) {
		Set<byte[]> values = zrangeByScore(key.getBytes(), min, max);
		Set<T> ret = null;
		if(values != null && !values.isEmpty()){
			ret = new HashSet<T>();
			for(byte[] value : values){
				ret.add((T)HessianSerializerUtils.deSerialize(value));
			}
			return ret;
		}else{
			return null;   
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrank(java.lang.String, byte[])
	 */
	public long zrank(String key, byte[] member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrank(key.getBytes(), member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrank(java.lang.String, java.lang.String)
	 */
	public long zrank(String key, String member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrank(key, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrankBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> long zrankBean(String key, T member) {
		return zrank(key, HessianSerializerUtils.serialize(member));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrange(byte[], long, long)
	 */
	public Set<byte[]> zrevrange(byte[] key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		try {
			value = jedis.zrevrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrange(java.lang.String, long, long)
	 */
	public Set<String> zrevrange(String key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		try {
			value = jedis.zrevrange(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrangeBean(java.lang.String, long, long)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> zrevrangeBean(String key,
			long start, long end) {
		Set<byte[]> values = zrevrange(key.getBytes(), start, end);
		Set<T> ret = null;
		if(values != null && !values.isEmpty()){
			ret = new HashSet<T>();
			for(byte[] value : values){
				ret.add((T)HessianSerializerUtils.deSerialize(value));
			}
			return ret;
		}else{
			return null;   
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zscore(java.lang.String, byte[])
	 */
	public Double zscore(String key, byte[] member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Double value = null;
		try {
			value = jedis.zscore(key.getBytes(), member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zscore(java.lang.String, java.lang.String)
	 */
	@Override
	public Double zscore(String key, String member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Double value = null;
		try {
			value = jedis.zscore(key, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zscoreBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> Double zscoreBean(String key, T member) {
		return zscore(key, HessianSerializerUtils.serialize(member));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrangeByScore(byte[], double, double)
	 */
	public Set<byte[]> zrevrangeByScore(byte[] key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<byte[]> value = null;
		try {
			value = jedis.zrevrangeByScore(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrangeByScore(java.lang.String, double, double)
	 */
	public Set<String> zrevrangeByScore(String key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Set<String> value = null;
		try {
			value = jedis.zrevrangeByScore(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrangeByScoreBean(java.lang.String, double, double)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> Set<T> zrevrangeByScoreBean(String key,
			double min, double max) {
		Set<byte[]> values = zrevrangeByScore(key.getBytes(), min, max);
		Set<T> ret = null;
		if(values != null && !values.isEmpty()){
			ret = new HashSet<T>();
			for(byte[] value : values){
				ret.add((T)HessianSerializerUtils.deSerialize(value));
			}
			return ret;
		}else{
			return null;   
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrank(java.lang.String, byte[])
	 */
	public long zrevrank(String key, byte[] member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrevrank(key.getBytes(), member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrank(java.lang.String, java.lang.String)
	 */
	public long zrevrank(String key, String member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrevrank(key, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zrevrankBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> long zrevrankBean(String key, T member) {
		return zrevrank(key, HessianSerializerUtils.serialize(member));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataReader#zgetPage(java.lang.String, int, int)
	 */
	public <T extends Serializable> List<T> zgetPage(String key, int pageNo,
			int pageSize) {
		List<T> ret = null;
		double min = (pageNo-1)* pageSize;
		double max = min + pageSize;
		Set<T> values = zrangeByScoreBean(key, min, max);
		if(values != null){
			ret = new ArrayList<T>();
			for(T t : values){
				ret.add(t);
			}
		}
		return ret;
	}
}
